package vn.com.epi.re.tools;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
import org.apache.hadoop.hbase.util.Bytes;

import vn.com.epi.re.ApplicationContext;
import vn.com.epi.re.table.ArticleTable;
import vn.com.epi.re.table.UserTable;

import com.google.common.io.Closeables;

/**
 * Utilities for table manipulation.
 * 
 */
public final class TableUtils {

	private static final int BULK_LIMIT = 10000;

	private TableUtils() {

	}

	private static final Log LOG = LogFactory.getLog(TableUtils.class);

	public static void deleteLastChar(StringBuilder sb) {
		if (sb.length() <= 0) {
			return;
		}
		sb.deleteCharAt(sb.length() - 1);
	}

	public static void truncateTable(Configuration conf, byte[] tableName) throws IOException {
		LOG.info("Started truncating table " + Bytes.toString(tableName));
		HBaseAdmin admin = new HBaseAdmin(conf);
		try {
			truncateTable(admin, tableName);
		}
		finally {
			Closeables.closeQuietly(admin);
		}
		LOG.info("Finished truncating table " + Bytes.toString(tableName));
	}

	public static void truncateTable(HBaseAdmin admin, byte[] tableName) throws IOException {
		HTableDescriptor descriptor = admin.getTableDescriptor(tableName);
		admin.disableTable(tableName);
		admin.deleteTable(tableName);
		admin.createTable(descriptor);
	}

	/**
	 * <p>
	 * <strong>Notice:</strong> We shouldn't create table in this function
	 * because whether one uses table pool or not may affect dramatically the
	 * performance. Let the user decides her strategy.
	 * </p>
	 * 
	 * @param table
	 * @param family
	 * @param column
	 * @throws IOException
	 */
	public static void deleteColumn(HTableInterface table, byte[] family, byte[] column) throws IOException {
		LOG.info("Start deleting column " + Bytes.toString(column) + " from table " + Bytes.toString(table.getTableName()));
		Scan scan = new Scan();
		scan.setFilter(new KeyOnlyFilter());
		ResultScanner rs = table.getScanner(scan);
		List<Delete> deleteList = new ArrayList<Delete>(BULK_LIMIT);
		try {
			for (Result result = rs.next(); result != null; result = rs.next()) {
				Delete delete = new Delete(result.getRow());
				delete.deleteColumns(family, column);
				deleteList.add(delete);
				if (deleteList.size() == BULK_LIMIT) {
					table.delete(deleteList);
					deleteList.clear();
				}
			}
			table.delete(deleteList);
		}
		finally {
			rs.close();
		}
		LOG.info("Finish deleting column " + Bytes.toString(column) + " from table " + Bytes.toString(table.getTableName()));
	}

	public static void deleteColumn(HTableInterface table, byte[] family, byte[] column, long timestamp) throws IOException {
		LOG.info("Start deleting column " + Bytes.toString(column) + " from table " + Bytes.toString(table.getTableName()) + " of which timestamp is less than " + timestamp);
		Scan scan = new Scan();
		scan.setFilter(new KeyOnlyFilter());
		ResultScanner rs = table.getScanner(scan);
		List<Delete> deleteList = new ArrayList<Delete>(BULK_LIMIT);
		try {
			for (Result result = rs.next(); result != null; result = rs.next()) {
				Delete delete = new Delete(result.getRow());
				delete.deleteColumns(family, column, timestamp);
				deleteList.add(delete);
				if (deleteList.size() == BULK_LIMIT) {
					table.delete(deleteList);
					deleteList.clear();
				}
			}
			table.delete(deleteList);
		}
		finally {
			rs.close();
		}
		LOG.info("Finish deleting column " + Bytes.toString(column) + " from table " + Bytes.toString(table.getTableName()) + " of which timestamp is less than " + timestamp);
	}

	public static void deleteRow(HTableInterface table, long timestamp) throws IOException {
		LOG.info("Start deleting all rows from table " + Bytes.toString(table.getTableName()) + " of which timestamp is less than " + timestamp);
		Scan scan = new Scan();
		scan.setTimeRange(0, timestamp);
		scan.setFilter(new KeyOnlyFilter());
		ResultScanner rs = table.getScanner(scan);
		List<Delete> deleteList = new ArrayList<Delete>(BULK_LIMIT);
		try {
			for (Result result = rs.next(); result != null; result = rs.next()) {
				Delete delete = new Delete(result.getRow(), timestamp, null);
				deleteList.add(delete);
				if (deleteList.size() == BULK_LIMIT) {
					table.delete(deleteList);
					deleteList.clear();
				}
			}
			table.delete(deleteList);
		}
		finally {
			rs.close();
		}
		LOG.info("Finish all rows from table " + Bytes.toString(table.getTableName()) + " of which timestamp is less than " + timestamp);
	}

	/**
	 * Split a comma-separated list of doubles. Default to 0 in case of wrong
	 * number format.
	 * 
	 * @param string
	 * @return
	 */
	public static double[] splitDoubles(String string) {
		String[] chunks = string.split(",");
		double[] doubles = new double[chunks.length];
		for (int i = 0; i < chunks.length; i++) {
			try {
				doubles[i] = Double.parseDouble(chunks[i]);
			}
			catch (NumberFormatException e) {
				doubles[i] = 0;
			}
		}
		return doubles;
	}

	/**
	 * Split a comma-separated list of doubles. Default to 0 in case of wrong
	 * number format.
	 * 
	 * @param string
	 * @return
	 */
	public static long[] splitLongs(String string) {
		String[] chunks = string.split(",");
		long[] longs = new long[chunks.length];
		for (int i = 0; i < chunks.length; i++) {
			try {
				longs[i] = Long.parseLong(chunks[i]);
			}
			catch (NumberFormatException e) {
				longs[i] = 0;
			}
		}
		return longs;
	}

	/**
	 * Close a table in a null-safe way and optionally suspending exceptions (to
	 * be used in finally blocks). See also {@link HTableWrapper} class that
	 * implements Closeable interface and
	 * {@link ApplicationContext#getTable(String)} method that returns such a
	 * wrapper.
	 * 
	 * @see Closeables#close(Closeable, boolean)
	 * @param table
	 * @param swallowIOException
	 * @throws IOException
	 */
	public static void close(HTableInterface table, boolean swallowIOException) throws IOException {
		if (table == null) {
			return;
		}
		try {
			table.close();
		}
		catch (IOException e) {
			if (swallowIOException) {
				LOG.warn("IOException thrown while closing Closeable.", e);
			}
			else {
				throw e;
			}
		}
	}

	/**
	 * Close a table without throwing any exception - to be used in finally
	 * blocks. See also {@link HTableWrapper} class that implements Closeable
	 * interface and {@link ApplicationContext#getTable(String)} method that
	 * returns such a wrapper.
	 * 
	 * @see Closeables#closeQuietly(Closeable)
	 * @param table
	 */
	public static void closeQuietly(HTableInterface table) {
		try {
			close(table, true);
		}
		catch (IOException e) {
			LOG.fatal("IOException should not have been thrown.", e);
		}
	}

	public static void createTable(HBaseAdmin admin, byte[] name, byte[]... families) throws IOException {
		if (admin.tableExists(name)) {
			LOG.info("Table already exists: " + Bytes.toString(name));
			return;
		}
		HTableDescriptor descriptor = new HTableDescriptor(name);
		for (int i = 0; i < families.length; i++) {
			descriptor.addFamily(new HColumnDescriptor(families[i]));
		}
		admin.createTable(descriptor);
	}

	public static void createTablesIfNotExist(Configuration configuration) throws IOException {
		HBaseAdmin admin = null;
		try {
			admin = new HBaseAdmin(configuration);
			createTable(admin, UserTable.NAME_BYTES, UserTable.FAM_MAIN);
			createTable(admin, ArticleTable.NAME_BYTES, ArticleTable.FAM_MAIN);
		}
		finally {
			if (admin != null) {
				admin.close();
			}
		}
	}
}
